package com.sfx.messagequeue.mqServer;

import com.sfx.messagequeue.common.consumer.Consumer;
import com.sfx.messagequeue.common.exception.MQException;
import com.sfx.messagequeue.mqServer.core.*;
import com.sfx.messagequeue.mqServer.datacenter.ConsumerManager;
import com.sfx.messagequeue.mqServer.datacenter.DiskDataCenter;
import com.sfx.messagequeue.mqServer.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * Description: 使用虚拟主机来进行管理 交换机,队列,绑定
 * 另外 VirtualHost把前面我们抛出的异常,全在这一层进行处理
 * User: sfx
 * Date: 2023-08-05
 * Time: 22:49
 */
public class VirtualHost {

    private final String virtualHostName;//虚拟机的唯一标识
    //内存数据管理
    private final MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    //磁盘数据管理
    private final DiskDataCenter diskDataCenter = new DiskDataCenter();
    //路由规则的设定
    private final Router router = new Router();
    //交换机锁对象
    private final Object EXCHANGE_LOCKER = new Object();
    //队列的锁对象
    private final Object QUEUE_LOCKER = new Object();
    // 消费者
    private final ConsumerManager consumerManager = new ConsumerManager(this);


    //初始化方法
    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;
        //初始化数据库表..
        diskDataCenter.init();
        try {
            //加载磁盘数据到内存
            memoryDataCenter.recovery(diskDataCenter);
        } catch (MQException | IOException | ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    /**
     * 实现exchangeDeclare方法,目的是创建一个交换机
     * 如果交换机存在,则不创建,否则 创建, 创建成功返回true,否则返回false.
     *
     * @param exchangeName 交换机名字
     * @param type         交换机类型
     * @param durable      是否持久化
     * @param autoDelete   是否自动删除
     * @param arguments    结构化参数,实现额外的功能
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType type, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) {
        //1.为了实现不同虚拟主机的隔离(每一个虚拟主机都管理着自己的交换机,队列,绑定),给交换机重命名
        exchangeName = getNewName(exchangeName);
        try {
            synchronized (EXCHANGE_LOCKER) {
                //2.交换机存在,则不创建
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if (exchange != null) {
                    System.out.println("[VirtualHost] 交换机已经存在 exchangeName=" + exchangeName);
                    return true;
                }
                //3.交换机不存在,则创建
                exchange = new Exchange(exchangeName, durable, autoDelete, type, arguments);
                //这里实现先将交换机保存到磁盘,在保存到内存,因为磁盘的速度比较慢,还有可能出现各种的异常
                //如果先写内存,如果成功了,但是写磁盘失败了,还需要删除内存中的数据
                //4.将交换机保存到磁盘中,持久化才存储
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                //5.交换机保存到内存中
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建成功 exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机创建失败 exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除指定exchangeName的交换机
     *
     * @param exchangeName 交换机名字
     * @return 是否删除交换机成功
     */
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = getNewName(exchangeName);
        try {
            synchronized (EXCHANGE_LOCKER) {
                //1.获取交换机
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                //2.交换机不存在,则抛出异常
                if (exchange == null) {
                    throw new MQException("[VirtualHost] 交换机不存在 exchangeName=" + exchangeName);
                }
                //3.交换机存在,则进行删除
                memoryDataCenter.deleteExchange(exchange);
                //4.是否持久化决定删除硬盘
                if (exchange.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                System.out.println("[VirtualHost] 交换机删除成功 exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败 exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 创建队列
     *
     * @param queueName  队列的名字
     * @param durable    是否支持持久化
     * @param autoDelete 自动删除
     * @param exclusive  是否独占
     * @param arguments  一些额外的参数,可以添加额外的功能
     * @return 创建队列是否成功
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean autoDelete,
                                boolean exclusive, Map<String, Object> arguments) {
        queueName = getNewName(queueName);
        try {
            synchronized (QUEUE_LOCKER) {
                //1.队列是否存在,存在则不创建,返回
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue != null) {
                    System.out.println("[VirtualHost] 队列已经存在 queueName=" + queueName);
                    return true;
                }
                //2.队列不存在,则创建队列
                queue = new MSGQueue(queueName, durable, exclusive, autoDelete, arguments);
                //3.保存到磁盘
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                //4.保存到内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 保存队列成功 queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 保存队列失败 queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除队列
     *
     * @param queueName 队列名字
     */
    public boolean queueDelete(String queueName) {
        queueName = getNewName(queueName);
        try {
            synchronized (QUEUE_LOCKER) {
                //1.得到队列
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                //2.队列不存在,则抛出异常
                if (queue == null) {
                    throw new MQException("[VirtualHost] 队列不存在 queueName=" + queueName);
                }
                //3.队列存在,则删除队列
                memoryDataCenter.deleteQueue(queue);
                //4.看是否持久化,决定删除硬盘上的队列
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                System.out.println("[VirtualHost] 队列删除成功 queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 队列删除失败 queueName=" + queueName);
            return false;
        }
    }

    private String getNewName(String name) {
        return virtualHostName + name;
    }

    /**
     * 创建绑定
     *
     * @param exchangeName 交换机的名字
     * @param queueName    队列的名字
     * @param bindingKey   bindingKye
     * @return
     */
    public boolean queueBind(String exchangeName, String queueName, String bindingKey) {
        queueName = getNewName(queueName);
        exchangeName = getNewName(exchangeName);
        try {
            synchronized (EXCHANGE_LOCKER) {
                synchronized (QUEUE_LOCKER) {
                    //1.绑定是否存在,绑定存在,则不创建
                    Binding binding = memoryDataCenter.getBinding(queueName, exchangeName);
                    if (binding != null) {
                        throw new MQException("[VirtualHost] 绑定存在 queueName=" + queueName
                                + " exchangeName=" + exchangeName);
                    }
                    //2.检查bindingKey是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MQException("[VirtualHost] bindingKey不合法  bindingKey : " + bindingKey);
                    }

                    //3.检查交换机,队列是否存在
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MQException("[VirtualHost] 交换机不存在,绑定无法创建  exchangeName=" + exchangeName);
                    }

                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MQException("[VirtualHost] 交换机不存在,绑定无法创建  queueName=" + queueName);
                    }
                    //4.绑定不存在,则创建
                    binding = new Binding();
                    binding.setBindingKey(bindingKey);
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    //5.交换机,队列是否支持持久化到硬盘
                    if (exchange.isDurable() && queue.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    //6.保存绑定到内存
                    memoryDataCenter.insertBinding(binding);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 创建绑定失败  queueName=" + queueName
                    + " exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除绑定
     *
     * @param queueName    队列名字
     * @param exchangeName 交换机名字
     * @return 返回是否删除绑定成功
     */
    public boolean queueUnbind(String queueName, String exchangeName) {
        queueName = getNewName(queueName);
        exchangeName = getNewName(exchangeName);
        try {
            synchronized (EXCHANGE_LOCKER) {
                synchronized (QUEUE_LOCKER) {
                    //1.检查绑定是否存在,不存在,则抛出异常
                    Binding binding = memoryDataCenter.getBinding(queueName, exchangeName);
                    if (binding == null) {
                        throw new MQException("[VirtualHost] 绑定不存在无法删除 queueName=" + queueName
                                + " exchangeName=" + exchangeName);
                    }
                    //2.存在则直接删除
                    diskDataCenter.deleteBinding(binding);
                    memoryDataCenter.deleteBinding(binding);
                }
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 删除绑定失败 queueName=" + queueName
                    + " exchangeName=" + exchangeName);
            return false;
        }
    }

    /**
     * 发送消息,将消息发送给交换机 or 队列
     *
     * @param exchangeName    交换机名字
     * @param basicProperties 消息的属性
     * @param routingKey      路由键
     * @return 返回 发送消息是否成功
     */
    public boolean basicPublish(String exchangeName,
                                BasicProperties basicProperties,
                                String routingKey,
                                byte[] body) {
        exchangeName = getNewName(exchangeName);

        try {
            //1.校验routingKey是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MQException("[VirtualHost] routingKey不合法 routingKey=" + routingKey);
            }
            //2.校验交换机是否存在
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MQException("[VirtualHost] 交换机不存在 exchangeName=" + exchangeName);
            }
            //3.检查交换机类型
            ExchangeType type = exchange.getType();
            if (type == ExchangeType.DIRECT) {
                //直接交换机
                // routingKey就是队列名,我们直接发送给这个队列即可
                //3.1 得到队列名字
                String queueName = getNewName(routingKey);
                //3.2 根据队列名得到队列
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MQException("[VirtualHost] 队列不存在(直接交换机) queueName=" + queueName);
                }
                //3.3 构造消息
                Message message = Message.createMessageWithId(basicProperties, routingKey, body);
                //3.4 发送消息
                sendMessage(queue, message);
                System.out.println("[VirtualHost] 直接交换机发送消息成功 queueName=" + queueName);
            } else {
                //扇出交换机 or 主题交换机
                // 扇出交换机 和 主题交换机都与绑定有关系
                // 1. 扇出交换机是给和给交换机绑定的所有队列全部发送消息
                // 2. 主题交换机是给能够对上暗号的(routingKey和bindingKey进行匹配)才能发送

                //3.1 遍历交换机所有的绑定
                //3.1.1 得到交换机上的所有绑定
                //               queueName binding
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                //3.1.2 遍历绑定
                for (Map.Entry<String, Binding> bindingEntry : bindingsMap.entrySet()) {
                    String queueName = getNewName(bindingEntry.getKey());
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        System.out.println("[VirtualHost] 队列不存在(扇出和主题交换机) queueName=" + queueName);
                        continue;
                    }
                    Message message = Message.createMessageWithId(basicProperties, routingKey, body);
                    //3.2 检查如果是扇出交换机,那么直接发送
                    //3.3 如果是主题交换机,必须routingKey和bindingKey匹配上才可以
                    // 这两个路由规则书写在router类中编写代码
                    Binding binding = bindingEntry.getValue();
                    if (!router.route(exchange.getType(), binding, message)) {
                        //如果路由规则不合法,就不用发送消息
                        System.out.println("[VirtualHost] sendMessage 发送消息时 路由规则不合法");
                        throw new MQException("[VirtualHost] 路由规则不合法 ");
                    }
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 发送消息失败");
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 发送消息-> 这里就是将消息进行保存,磁盘 + 内存
     * 如果需要保存磁盘deliverMode=2;
     *
     * @param queue   队列
     * @param message 消息
     */
    private void sendMessage(MSGQueue queue, Message message) throws InterruptedException {
        //内存保存一份
        memoryDataCenter.sendMessage(queue, message);
        //看是否需要内存保存
        if (message.getDeliverMode() == 2) {
            try {
                diskDataCenter.sendMessage(queue, message);
            } catch (MQException | IOException e) {
                System.out.println("[VirtualHost] 发送消息时硬盘保存失败");
                e.printStackTrace();
            }
        }
        // 这里发送完消息之后,还要通知消费者可以消费了
        consumerManager.notifyConsume(queue.getName());
    }


    /**
     * 主要实现订阅消息
     * 往队列中添加一个订阅者,收到消息后,把消息推送给对应的订阅者->消费者
     *
     * @param consumerTag 消费者的标识
     * @param queueName   要订阅消息的队列
     * @param autoAck     应答方式 为true代表自动应答(消息拿走,啥也不干)   false手动应答(需要调用basicAck方法)
     * @param consumer    回调函数, 这里主要处理当触发basicConsume方法的时候,需要触发回调函数consumer
     */
    public boolean basicConsume(String consumerTag,
                                String queueName,
                                boolean autoAck,
                                Consumer consumer) {
        //区分虚拟主机
        queueName = getNewName(queueName);
        try {
            //给队列添加订阅者,消息来了之后,并把消息推送给订阅者
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功 queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume 失败 queueName=" + queueName);
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 实现消息的确认
     *
     * @param queueName 队列
     * @param messageId 消息id
     * @return 返回确认是否成功
     */
    public boolean basicAck(String queueName, String messageId) {
        queueName = getNewName(queueName);
        try {
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MQException("[VirtualHost] basicAck 消息不存在 queueName=" + queueName + " massageId=" + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MQException("[VirtualHost] basicAck 队列不存在 queueName=" + queueName + " massageId=" + messageId);
            }
            // isAutoAck 为true 代表自动应答,如果回调不抛出异常,直接将消息删除
            // 1): 看消息是否持久化,如果持久化,从磁盘中删除
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            // 2): 消息从内存中删除
            memoryDataCenter.deleteMessage(message.getMessageId());
            // 3): 消息从WaitAckMap中删除
            memoryDataCenter.removeMessageWaitAck(queue.getName(), message);
            System.out.println("[VirtualHost] basicAck 成功! 消息被成功确认! queueName=" + queueName
                    + ", messageId=" + messageId);
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck 消息确认失败 queueName=" + queueName + " massageId=" + messageId);
            e.printStackTrace();
        }
        return false;
    }
}